/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.sss.client.SimpleJob;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Reducer;

public class KMeans {
  private static Class<? extends Mapper> mapperClass = KMeansMapper.class;
  private static Class<? extends Mapper> bulkMapperClass = KMeansBulkMapper.class;
  private static Class<? extends Reducer> combinerClass = KMeansReducer.class;
  // static Class<? extends Reducer> combinerClass = null;
  private static Class<? extends Reducer> reducerClass = KMeansReducer.class;

  public static final String CENTER_KEY = "kmeans.previous.centers";

  private static SssClient client = null;

  private static SimpleDateFormat dataFormatter = new SimpleDateFormat("yyyyMMddkkmmss");

  private static void printTimestamp(String what) {
    System.err.println("@ " + what  + " " + dataFormatter.format(new Date(System.currentTimeMillis())));
  }

  public static void usage(Options options) {
    System.out.println("java KMeans N group_id");
    HelpFormatter f = new HelpFormatter();
    f.printHelp("KMeans", options);
    System.exit(-1);
  }

  public static GroupID run(GroupID input, HashMap<Integer, Center> centers, boolean bulk)
      throws Exception {
    SimpleJob job = new SimpleJob(client);
    job.getConfiguration().setObject(CENTER_KEY, centers);

    if (bulk) {
      job.setMapperClass(bulkMapperClass);
    }
    else {
      job.setMapperClass(mapperClass);
    }
    job.setCombinerClass(combinerClass);
    job.setReducerClass(reducerClass);
    job.setInputGID(input);

    return job.execute();
  }

  public static GroupID loop(GroupID input, 
                             HashMap<Integer, Center> centers, 
                             boolean bulk, 
                             int numExecution) throws Exception {
    KMeansResultPrinter reader = new KMeansResultPrinter(client);
    GroupID out = null;
    int n = 0;
    while ((numExecution <= 0) || n < numExecution) {
      printTimestamp("job_start");

      long start = System.currentTimeMillis();
      out = run(input, centers, bulk);
      long end = System.currentTimeMillis();
      long elapsed = end - start;
      System.out.println("elapsed a loop: " + Long.toString(elapsed) + " ms.");
      printTimestamp("job_done");
      HashMap<Integer, Center> newCenters = reader.readMap(out);
      if (Center.compare(centers, newCenters)) {
        break;
      }
      centers = newCenters;
      System.err.println(centers);
      n++;
    }
    assert out != null;
    return out;
  }

  /**
   * Usage: java KMeans groupid
   */
  public static void main(String[] args) throws Exception {
    printTimestamp("kmeans_start");
    boolean bulk = false;
    int numExecution = 0;

    client = new SssClient(args);

    Options options = new Options();
    options.addOption("b", "bulk", false, "use bulk input.");
    options.addOption("n", "number", true, "number of executions.");

    CommandLineParser parser = new GnuParser();
    CommandLine cl = parser.parse(options, client.getArgs());

    args = cl.getArgs();
    if (args.length < 2) {
      usage(options);
    }
    int numCenters = Integer.parseInt(args[0]);
    GroupID input = GroupID.createFromString(args[1]);

    bulk = cl.hasOption("b");
    if (cl.hasOption("n")) {
      numExecution = Integer.parseInt(cl.getOptionValue("n"));
    }
    if (numCenters <= 0) {
      usage(options);
    }

    HashMap<Integer, Center> centers = new HashMap<Integer, Center>();
    for (int i = 0; i < numCenters; i++) {
      centers.put(i, Center.getRandom());
    }

    long start = System.currentTimeMillis();
    GroupID out = loop(input, centers, bulk, numExecution);
    long end = System.currentTimeMillis();
    long elapsed = end - start;

    System.out.println("output data deployed - " + out.toString());
    System.out.println("elapsed " + Long.toString(elapsed) + " ms.");

    printTimestamp("kmeans_done");
  }
}
